package de.preissler.createview;

import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;

import me.prettyprint.cassandra.serializers.DoubleSerializer;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.HSuperColumn;
import me.prettyprint.hector.api.beans.Row;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;
import me.prettyprint.hector.api.query.RangeSlicesQuery;
import me.prettyprint.hector.api.query.SubColumnQuery;
import me.prettyprint.hector.api.query.SuperSliceQuery;

public class DAXPreCalc
{
	private static final String DAX = "DAX";
	private static final String DAX_VALUE = "value";
	private static final String CLOSE = "close";
	private static final String KT = "Kt";
	private static final String BASE = "base";
	private static final String PI0 = "price at zero";
	private static final String QI0 = "#shares at zero";
	private static final String QIT = "#shares at t";
	private static final String FFI = "free float";
	private static final String CI = "correction";
	private static final String INDEX_ISIN = "DE0008469008";
	
	private Keyspace keyspace;
	
	public DAXPreCalc(Keyspace keyspace)
	{
		this.keyspace = keyspace;
	}
	
	private double getValue(Keyspace keyspace, String scf, String key, String scn, String cn)
	{
		SubColumnQuery<String, String, String, Double> query = HFactory.createSubColumnQuery(keyspace, MyUtil.SS, MyUtil.SS, MyUtil.SS, MyUtil.DS);
		query.setColumnFamily(scf);
		query.setKey(key);
		query.setSuperColumn(scn);
		query.setColumn(cn);
		return query.execute().get().getValue();
	}
	
	/*
	 * Holt die ISINS
	 */
	private List<String> getISINs(Keyspace keyspace, String date)
	{
		SuperSliceQuery<String, String, String, Double> query = HFactory.createSuperSliceQuery(keyspace, StringSerializer.get(),
				StringSerializer.get(), StringSerializer.get(), DoubleSerializer.get());
		query.setColumnFamily(DAX);
		query.setKey(date);
		query.setRange(null, null, false, Integer.MAX_VALUE);
		List<HSuperColumn<String, String, Double>> result = query.execute().get().getSuperColumns();
		List<String> isins = new LinkedList<String>();
		
		for(HSuperColumn<String, String, Double> sc : result)
			isins.add(sc.getName());
		
		return isins;
	}
	
	/*
	 * Holt alle Schluessel raus
	 */
	private List<String> getKeys(Keyspace keyspace, String nameOfColumnFamily)
	{
		List<List<String>> listsOfISINs = new LinkedList<List<String>>();
		
		String start = null;
		String end = null;
		boolean firstLoop = true;
		//Max. werden nur 100 Schluessel zurueck gegeben, deswegen diese Schleife
		while(true)
		{
			List<String> isins = new LinkedList<String>();
			
	         RangeSlicesQuery<String, String, String> rangeSlicesQuery = 
	        	 HFactory.createRangeSlicesQuery(keyspace, StringSerializer.get(), StringSerializer.get(), StringSerializer.get());
		     rangeSlicesQuery.setColumnFamily(nameOfColumnFamily);
		     rangeSlicesQuery.setReturnKeysOnly();
		     rangeSlicesQuery.setKeys(start, end);
		     List<Row<String, String, String>> result = rangeSlicesQuery.execute().get().getList();
		     
		     for(int i = 0; i < result.size(); i++)
		     {
		    	 if((!firstLoop) && (i == 0))
		    	 {
		    		 
		    	 }
		    	 else
		    		 isins.add(result.get(i).getKey());
		    		 
		     }
		     
		     if(isins.isEmpty())
		    	 break;
		     else
		     {
		    	 start = isins.get(isins.size()-1);
		    	 listsOfISINs.add(isins);
		    	 firstLoop = false;
		     }
		}
		
		List<String> isins = new LinkedList<String>();
		for(List<String> subList : listsOfISINs)
		{
			isins.addAll(subList);
		}
	     
		return isins;
	}
	
	/*
	 * Erstellt den Index
	 */
	@SuppressWarnings("unchecked")
	public void createIndex() throws Exception
	{
		//alle verfuegbaren ISINS und Dates holen
		List<String> dates = getKeys(keyspace, Main.SCF_RVT_JOIN);
		Mutator<String> mutator = HFactory.createMutator(keyspace, MyUtil.SS);
		int i = 1;
		
		//Fuer jedes Datum
		for(String date : dates)
		{
			MyUtil.trace("Working on " + date + " (" + i + "/" + dates.size() + ")");
			List<String> isins = getISINs(keyspace, date);
			double divident = 0.0;
			double divisior = 0.0;
			double k = 1.0;
			double base = 1000.0;
			
			for(String isin : isins)
			{
				//Wenn es der DAX ist brauchen wir andere Werte
				if(!isin.equals(INDEX_ISIN))
				{
					//hier werden die Werte geholt
					double close = getValue(keyspace, Main.SCF_RVT_JOIN, date, isin, CLOSE);
					double pi0 = getValue(keyspace, Main.SCF_RVT_JOIN, date, isin, PI0);
					if(pi0 == -1.0)
						pi0 = 1.0;
					double qi0 = getValue(keyspace, Main.SCF_RVT_JOIN, date, isin, QI0);
					if(qi0 == -1.0)
						qi0 = 1.0;
					double qit = getValue(keyspace, Main.SCF_RVT_JOIN, date, isin, QIT);
					if(qit == -1.0)
						qit = 1.0;
					double ffit = getValue(keyspace, Main.SCF_RVT_JOIN, date, isin, FFI);
					if(ffit == -1.0)
						ffit = 1.0;
					double ci = getValue(keyspace, Main.SCF_RVT_JOIN, date, isin, CI);
					if(ci == -1.0)
						ci = 1.0;
					
					divident = divident + (close*qit*ffit*ci);
					divisior = divisior + (pi0 * qi0);
					

				}
				else
				{
					//DAX Werte auslesen
					k = getValue(keyspace, Main.SCF_RVT_JOIN, date, isin, KT);
					base = getValue(keyspace, Main.SCF_RVT_JOIN, date, isin, BASE);
				}
			}
			
			double quotient = divident / divisior;
			double result = k * quotient * base;
			
			mutator.insert(date, Main.SCF_RVT_JOIN, HFactory.createSuperColumn(INDEX_ISIN, Arrays.asList
					(HFactory.createColumn(DAX_VALUE, result, MyUtil.SS, MyUtil.DS)), MyUtil.SS, MyUtil.SS, MyUtil.DS));
			
			i++;
		}
		
	}

}
